/**
 * @Author: Routz
 * @since: 2020/10/27 23:41
 */

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * <p>
 * A simple example program to use DataMonitor to start and
 * stop executables based on a znode. The program watches the
 * specified znode and saves the data that corresponds to the
 * znode in the filesystem. It also starts the specified program
 * with the specified arguments when the znode exists and kills
 * the program if the znode goes away.
 * </p>
 * maintains the ZooKeeper connection
 * contains the main thread and contains the execution logic
 *
 * The Watcher interface is defined by the ZooKeeper Java API.
 * ZooKeeper uses it to communicate back to its container.
 * It supports only one method, process(),
 * and ZooKeeper uses it to communicates generic events that the main thread would be interested in,
 * such as the state of the ZooKeeper connection or the ZooKeeper session.
 * The Executor in this example simply forwards those events down to the DataMonitor to decide what to do with them.
 *
 * It does this simply to illustrate the point that, by convention,
 * the Executor or some Executor-like object "owns" the ZooKeeper connection,
 * but it is free to delegate the events to other events to other objects.
 * It also uses this as the default channel on which to fire watch events. (More on this later.)
 */
public class Executor
        implements Watcher, Runnable, DataMonitor.DataMonitorListener {
    String znode;
    DataMonitor dm;
    ZooKeeper zk;
    String filename;
    String exec[];
    Process child;

    /**
     *
     * the Executor passes a reference to itself as the Watcher argument in the ZooKeeper constructor.
     * It also passes a reference to itself as DataMonitorListener argument to the DataMonitor constructor.
     * @param hostPort
     * @param znode
     * @param filename
     * @param exec
     * @throws KeeperException
     * @throws IOException
     */
    public Executor(String hostPort, String znode, String filename,
                    String exec[]) throws KeeperException, IOException {
        // 存储data
        this.filename = filename;
        // 参数
        this.exec = exec;
        // 当前对象作为watcher
        zk = new ZooKeeper(hostPort, 3000, this);
        // 当前对象作为listener
        dm = new DataMonitor(zk, znode, null, this);
    }

    /**
     * the Executor's job is to start and stop the executable whose name you pass in on the command line.
     * It does this in response to events fired by the ZooKeeper object.
     * @param args
     */
    public static void main(String[] args) {
        if (args.length < 4) {
            System.err
                    .println("USAGE: Executor hostPort znode filename program [args ...]");
            System.exit(2);
        }
        String hostPort = args[0];
        String znode = args[1];
        String filename = args[2];
        String[] exec = new String[args.length - 3];
        // 将filename后的参数拷贝到exec
        System.arraycopy(args, 3, exec, 0, exec.length);
        try {
            new Executor(hostPort, znode, filename, exec).run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /***************************************************************************
     * We do process any events ourselves, we just need to forward them on.
     *
     * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.WatchedEvent)
     */
    @Override
    public void process(WatchedEvent event) {
        dm.process(event);
    }

    @Override
    public void run() {
        try {
            synchronized (this) {
                while (!dm.dead) {
                    wait();
                }
            }
        } catch (InterruptedException e) {
        }
    }

    @Override
    public void closing(int rc) {
        synchronized (this) {
            notifyAll();
        }
    }

    static class StreamWriter extends Thread {
        OutputStream os;

        InputStream is;

        StreamWriter(InputStream is, OutputStream os) {
            this.is = is;
            this.os = os;
            start();
        }

        public void run() {
            byte b[] = new byte[80];
            int rc;
            try {
                while ((rc = is.read(b)) > 0) {
                    os.write(b, 0, rc);
                }
            } catch (IOException e) {
            }

        }
    }

    @Override
    public void exists(byte[] data) {
        if (data == null) {
            if (child != null) {
                System.out.println("Killing process");
                child.destroy();
                try {
                    child.waitFor();
                } catch (InterruptedException e) {
                }
            }
            child = null;
        } else {
            if (child != null) {
                System.out.println("Stopping child");
                child.destroy();
                try {
                    child.waitFor();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                // 写出data到文件
                FileOutputStream fos = new FileOutputStream(filename);
                fos.write(data);
                fos.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                System.out.println("Starting child");
                // 初始化child 执行系统指令
                child = Runtime.getRuntime().exec(exec);
                // 把child子进程的输出打印出来
                new StreamWriter(child.getInputStream(), System.out);
                new StreamWriter(child.getErrorStream(), System.err);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
